package com.rock.job;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class FlinkCDCJob {

  @Value("${flink.mysql-cdc.hostname}")
  private String hostname;

  @Value("${flink.mysql-cdc.port}")
  private Integer port;

  @Value("${flink.mysql-cdc.database-name}")
  private String databaseName;

  @Value("${flink.mysql-cdc.table-name}")
  private String tableName;

  @Value("${flink.mysql-cdc.username}")
  private String username;

  @Value("${flink.mysql-cdc.password}")
  private String password;

  @Value("${flink.mysql-cdc.server-time-zone}")
  private String serverTimeZone;

  public void start(){
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
        .hostname(hostname)
        .port(port)
        .databaseList(databaseName)
        .tableList(databaseName + "." + tableName)
        .username(username)
        .password(password)
        .deserializer(new JsonDebeziumDeserializationSchema()) // 将 CDC 事件反序列化为 JSON
        .startupOptions(StartupOptions.latest()) // 从最新的 binlog 位置开始读取
        .build();

    // 解析JSON事件，提取操作类型（op）和数据（before/after）
    env.fromSource(mySqlSource, WatermarkStrategy.noWatermarks(), "MySQL Source")
        .map(FlinkCDCJob::processChangeEvent); // 自定义Sink（如写入Kafka、另一个MySQL等
    env.setRuntimeMode(RuntimeExecutionMode.STREAMING);

    try {
//      env.execute("MySQL Incremental Sync Job");
      env.executeAsync("MySQL Incremental Sync Job");
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

  private static String processChangeEvent(String event) {
    System.out.println("Received CDC event: " + event);
    return event;
  }

}